package org.codehaus.activemq.store.jdbm;

import java.io.File;
import java.io.IOException;
import java.util.Comparator;
import java.util.Properties;
import javax.jms.JMSException;
import jdbm.RecordManager;
import jdbm.RecordManagerFactory;
import jdbm.btree.BTree;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.AlreadyClosedException;
import org.codehaus.activemq.service.impl.PersistenceAdapterSupport;
import org.codehaus.activemq.store.MessageStore;
import org.codehaus.activemq.store.PreparedTransactionStore;
import org.codehaus.activemq.store.TopicMessageStore;
import org.codehaus.activemq.util.DefaultComparator;
import org.codehaus.activemq.util.JMSExceptionHelper;

public class JdbmPersistenceAdapter extends PersistenceAdapterSupport
{
  private static final Log log = LogFactory.getLog(JdbmPersistenceAdapter.class);
  private RecordManager manager;
  private File directory = new File("ActiveMQ");
  private Properties properties;

  public static JdbmPersistenceAdapter newInstance(File directory)
    throws JMSException
  {
    return new JdbmPersistenceAdapter(directory);
  }

  public JdbmPersistenceAdapter()
  {
  }

  public JdbmPersistenceAdapter(File directory) {
    this.directory = directory;
  }

  public JdbmPersistenceAdapter(RecordManager manager) {
    this.manager = manager;
  }

  public MessageStore createQueueMessageStore(String destinationName) throws JMSException {
    try {
      BTree messageDb = createDatabase("Queue_" + destinationName);
      BTree sequenceDb = createDatabase("Sequence_Queue_" + destinationName);
      JdbmMessageStore messageStore = new JdbmMessageStore(messageDb, sequenceDb);
      return messageStore;
    } catch (IOException e) {
    	   throw JMSExceptionHelper.newJMSException("Failed to create a QueueMessageContainer for destination: " + destinationName + ". Reason: " + e, e);

    }
   }

  public TopicMessageStore createTopicMessageStore(String destinationName) throws JMSException
  {
    try {
      BTree messageDb = createDatabase("Topic_" + destinationName);
      BTree sequenceDb = createDatabase("Sequence_Topic_" + destinationName);
      BTree consumerAckDb = createDatabase("Consumer_Acks_Topic_" + destinationName);
      BTree subscriberDb = createDatabase("Subscriber_" + destinationName);
      BTree messageCountDb = createDatabase("MessageCount_Topic_" + destinationName);
      JdbmTopicMessageStore messageStore = new JdbmTopicMessageStore(messageDb, sequenceDb, consumerAckDb, subscriberDb, messageCountDb);
      return messageStore;
    } catch (IOException e) {
    	   throw JMSExceptionHelper.newJMSException("Failed to create a TopicMessageContainer for destination: " + destinationName + ". Reason: " + e, e);

    }
   }

  public PreparedTransactionStore createPreparedTransactionStore() throws JMSException
  {
    try {
      return new JdbmPreparedTransactionStore(createDatabase("XaPrepareTxnDb"));
    } catch (IOException e) {
    	   throw JMSExceptionHelper.newJMSException("Could not create XA Prepare Transaction Database. Reason: " + e, e);
    	   
    }
  }

  public void beginTransaction()
  {
  }

  public void commitTransaction() throws JMSException {
    try {
      this.manager.commit();
    }
    catch (IOException e) {
      throw JMSExceptionHelper.newJMSException("Could not commit transaction. Reason: " + e, e);
    }
  }

  public void rollbackTransaction() {
    try {
      this.manager.rollback();
    }
    catch (IOException e) {
      log.error("Could not rollback transaction. Reason: " + e, e);
    }
  }

  public void start() throws JMSException {
    if (this.manager == null) {
      this.directory.mkdirs();

      log.info("Creating JDBM based message store in directory: " + this.directory.getAbsolutePath());
      try
      {
        String name = this.directory.getAbsolutePath() + "/Store";
        if (this.properties != null) {
          this.manager = RecordManagerFactory.createRecordManager(name, this.properties);
        }
        else
          this.manager = RecordManagerFactory.createRecordManager(name);
      }
      catch (IOException e)
      {
        throw JMSExceptionHelper.newJMSException("Failed to create JDBM persistent store at directory: " + this.directory + ". Reason: " + e, e);
      }
    }
  }

  public synchronized void stop() throws JMSException
  {
    if (this.manager != null)
      try {
        this.manager.close();
      }
      catch (IOException e) {
        throw JMSExceptionHelper.newJMSException("Failed to close PersistenceAdapter. Reason: " + e, e);
      }
      finally {
        this.manager = null;
      }
  }

  public RecordManager getManager()
  {
    return this.manager;
  }

  public void setManager(RecordManager manager) {
    this.manager = manager;
  }

  public File getDirectory() {
    return this.directory;
  }

  public void setDirectory(File directory) {
    this.directory = directory;
  }

  public synchronized BTree createDatabase(String name)
    throws IOException, AlreadyClosedException
  {
    if (this.manager == null) {
      throw new AlreadyClosedException("JDBM PersistenceAdapter");
    }

    long recid = this.manager.getNamedObject(name);
    BTree tree = null;
    if (recid != 0L) {
      tree = BTree.load(this.manager, recid);
    }
    else {
      Comparator comparator = new DefaultComparator();

      tree = BTree.createInstance(this.manager, comparator);
      this.manager.setNamedObject(name, tree.getRecid());
    }
    return tree;
  }
}